package com.iot.service.mqtt.service.rel;

import com.iot.service.mqtt.model.request.TlMqttPubRelReq;
import com.iot.service.mqtt.service.StoreStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @Author: hszhou
 * @Date: 2025/4/18 16:40
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.REDIS)
@Slf4j
public class RedisPubrelStoreServiceImpl implements IPubrelStoreService {

    private final ReactiveRedisTemplate<String,Object> reactiveRedisTemplate;

    private final static String PREFIX = "pubrel:";

    @Override
    public Mono<Boolean> savePubRelMessage(String clientId, Long messageId, TlMqttPubRelReq req) {
        return reactiveRedisTemplate.opsForHash().put(PREFIX+clientId, String.valueOf(messageId), req)
                .doOnError(e -> log.error("findPublishMessage byClientId failed for key: {}", PREFIX+clientId, e));
    }

    @Override
    public Mono<TlMqttPubRelReq> clearPubRelMessage(String clientId, Long messageId) {
        return reactiveRedisTemplate.opsForHash().remove(PREFIX+clientId, String.valueOf(messageId))
                .flatMap(value->findPubRelMessage(clientId,messageId))
                .doOnSuccess(e->{
                }).doOnError(e-> log.error("删除PUBREL【{}】的消息【{}】出现错误",clientId,messageId));
    }

    @Override
    public Mono<Boolean> clearPubRelMessage(String clientId) {
        return reactiveRedisTemplate.opsForHash().delete(PREFIX+clientId)
                .doOnError(e -> log.error("clearPubRelMessage byClientId failed for key: {}", PREFIX+clientId, e));
    }

    @Override
    public Mono<TlMqttPubRelReq> findPubRelMessage(String clientId, Long messageId) {
        return reactiveRedisTemplate.opsForHash().get(PREFIX+clientId, String.valueOf(messageId))
                .doOnError(e -> log.error("findPubRelMessage byClientId failed for key: {}", PREFIX+clientId, e))
                .cast(TlMqttPubRelReq.class);
    }

    @Override
    public Flux<TlMqttPubRelReq> findPubRelMessage(String clientId) {

        return reactiveRedisTemplate.opsForHash().keys(PREFIX+clientId)
                .doOnError(e -> log.error("findPubRelMessage byClientId failed for key: {}", PREFIX+clientId, e))
                .cast(TlMqttPubRelReq.class);
    }
}
